package org.weishe.pagerank;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class PeopleRankMain {

	/**
	 * 计算总条数
	 * 
	 * @throws Exception
	 */
	private static void preRun() throws Exception {
		String inputPath = "hdfs://192.168.15.10:9000/pagerank/people.txt";
		String outputPath = "hdfs://192.168.15.10:9000/pagerank/count";

		// 1、配置文件
		Configuration conf = new Configuration();
		conf.set("mapred.jar", "/Users/bill/Desktop/peoplerank.jar");

		Job job = Job.getInstance(conf);
		job.setJarByClass(PeopleRankMain.class);
		// 2、设置Job

		job.setJobName("PeopleRankPre");

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setMapperClass(PreMapper.class);
		job.setReducerClass(PreReducer.class);

		// 设置reducer的数量

		FileInputFormat.setInputPaths(job, new Path(inputPath));
		Path op = new Path(outputPath);

		FileOutputFormat.setOutputPath(job, op);

		boolean f = job.waitForCompletion(true);
		if (f) {
			System.out.println("pre job 成功执行");
		}
	}

	private static void setCount(Configuration conf) throws IOException {
		String countPath = "file:///pagerank/count/part-r-00000";
		Path op = new Path(countPath);
		FileSystem fs = FileSystem.get(conf);
		FSDataInputStream fis = fs.open(op);
		int count = fis.readInt();
		conf.set("sumcount", count + "");
		System.out.println("count:" + count);
		fis.close();
	}

	private static void run(int step, int count) throws Exception {

		String inputPath = "hdfs://192.168.15.10:9000/pagerank/output" + (step) + "/";
		if (step == 0) {
			inputPath = "hdfs://192.168.15.10:9000/pagerank/people.txt";
		}
		String outputPath = "hdfs://192.168.15.10:9000/pagerank/output" + (step + 1);

		// 1、配置文件
		Configuration conf = new Configuration();
		conf.set("mapred.jar", "/Users/bill/Desktop/peoplerank.jar");

		//setCount(conf);
		String countPath = "file:///pagerank/count/part-r-00000";
		Path op = new Path(countPath);
		FileSystem fs = FileSystem.get(conf);
		FSDataInputStream fis = fs.open(op);
		count = fis.readInt();
		conf.set("sumcount", count + "");
		System.out.println("count:" + count);
		fis.close();
		
		//conf.set("sumcount", count + "");
		Job job = Job.getInstance(conf);
		job.setJarByClass(PeopleRankMain.class);
		// 2、设置Job

		job.setJobName("PeopleRank step:" + step);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setMapperClass(PRMapper.class);
		job.setReducerClass(PRReducer.class);

		// 设置reducer的数量

		FileInputFormat.setInputPaths(job, new Path[] { new Path(inputPath) });
		Path op = new Path(outputPath);

		FileOutputFormat.setOutputPath(job, op);

		boolean f = job.waitForCompletion(true);
		if (f) {
			System.out.println("step:" + step + " job 成功执行");
		}
	}

	public static void main(String[] args) throws Exception {
		preRun();
		for (int i = 0; i < 1; i++) {
			run(i, 4);
		}

	}
}
